package com.ylkj.bus.device.report;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.hwl.redis.RedisConfig;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.device.onoffline.constant.OnOfflineMsgConstant;
import com.ylkj.bus.device.onoffline.model.OnOfflineDeviceEnty;
import com.ylkj.bus.device.report.constant.MsgKeyConstant;
import com.ylkj.bus.device.report.mode.DeviceMoDO;
import com.ylkj.bus.device.report.mode.ProductDO;
import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import com.ylkj.dao.RedisConfigInfo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.BeanListHandler;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.text.MessageFormat;
import java.util.*;


@Slf4j
public class DeviceOnlineStatusProcess extends ProcessFunction<MessageContent, MessageContent> {


    public static final String T_IOTMO_DEVICE_INIT_SQL = "SELECT device_code as deviceCode, product_id as productId FROM {0} WHERE 1=1 LIMIT ? OFFSET ?";
    public static final String T_IOTMO_PRODUCT_INIT_SQL = "SELECT product_id as productId, product_type as productType FROM {0} WHERE 1=1 LIMIT ? OFFSET ?";

    private final OutputTag<OnOfflineDeviceEnty> outTagStatus;
    private final RdsConfig rdsConfig;
    private final String tableName5;
    private final String tableName6;
    private final RedisConfigInfo redisConfig;
    private String rds5SQL;
    private String rds6SQL;
    private RdsUtils rdsUtils;
    private final Map<String, Long> iotpDevicecaches = new HashMap<>();
    private final Map<Long, String> iotProductcaches = new HashMap<>();
    private Jedis jedis;



    /**
     * 设备状态: 上线
     */
    public static final String ONLINE = "deviceOnline";

    /**
     * 设备状态: 下线
     */
    public static final String OFFLLINE = "deviceOffline";


    public DeviceOnlineStatusProcess(OutputTag<OnOfflineDeviceEnty> outTagStatus, RdsConfig rdsConfig,
                                     String rdsTableName5, String rdsTableName6,
                                     RedisConfigInfo redisConfig1) {
        this.outTagStatus = outTagStatus;
        this.rdsConfig = rdsConfig;
        this.tableName5 = rdsTableName5;
        this.tableName6 = rdsTableName6;
        this.redisConfig = redisConfig1;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        rdsUtils = new RdsUtils(rdsConfig);
        jedis = redisConfig.getJedis();
        rds5SQL = MessageFormat.format(T_IOTMO_DEVICE_INIT_SQL, tableName5);
        rds6SQL = MessageFormat.format(T_IOTMO_PRODUCT_INIT_SQL, tableName6);
        initDeviceCache();
        initProductCache();
    }


    @Override
    public void processElement(MessageContent msg, Context ctx, Collector<MessageContent> out) throws Exception {
        JSONObject body = JSONObject.parseObject(msg.getData());
        if(body.isEmpty()){
            log.error("原始数据信息上报的数据为空：{}",msg);
            return;
        }
        Map<String, Object> map = new HashMap<>();
        String notifyType = body.getString(MsgKeyConstant.NOTIFY_TYPE);
        String deviceCode = body.getString(MsgKeyConstant.DEVICE_IDENTIFIER);
        String eventTime = body.getString(MsgKeyConstant.EVENT_TIME);
        Long projectId = iotpDevicecaches.get(deviceCode);  // 从Integer改为Long
        String productType = iotProductcaches.get(projectId);  // 从Integer改为Long
        //long eventTimeLong = Long.parseLong(eventTime);
        DateTime date = DateUtil.parse(eventTime);
        if (StrUtil.isEmpty(deviceCode)) {
            log.error("原始数据信息上报的deviceCode为空：{}",msg);
            return;
        }
        if(!Objects.equals(notifyType,ONLINE) && !Objects.equals(notifyType,OFFLLINE)){
            log.error("原始数据信息上报的上下线值为空：{}",msg);
            return;
        }
        //将数据写入到Redis中
        handleDataToRedis(date,deviceCode,notifyType);

        Date now = DateUtil.date();
        OnOfflineDeviceEnty enty = new OnOfflineDeviceEnty();
        if(Objects.equals(notifyType,ONLINE)){
            enty.setRomaOnlineStatus(OnOfflineMsgConstant.DEVICE_ONLINE);
            enty.setAlarmCover("3");
            enty.setPersonnelHandler("");
            enty.setHandlingTime(date);
        }else {
            enty.setRomaOnlineStatus(OnOfflineMsgConstant.DEVICE_OFFLINE);
            enty.setAlarmTime(date);
            enty.setAlarmCover("1");
            map.put("lastOfficeTime", date.getTime());
            map.put("offLineTime", 0);
            // 将 map 转换为 JSON 字符串
            String mapAsJson = JSON.toJSONString(map);
            enty.setEventValue(mapAsJson);
        }
        enty.setLastOnlineStatusTime(date);
        enty.setUpdateTime(now);
        enty.setDeviceCode(deviceCode);
        enty.setEventType("last_offline");
        enty.setProductId(projectId);
        enty.setProductType(productType);
        enty.setNetCode("1001");
        enty.setAlertLevel("2");
        enty.setBlankingTool("DEFECT_ELIMINATION");
        enty.setMessageSuggestion("前往现场查看离线原因");
        enty.setAlertDescription(StrUtil.format("设备当前为离线状态",StrUtil.SPACE));
        enty.setAlertSource("固定检测");
        enty.setAlertDimensions("1");
        ctx.output(outTagStatus,enty);
    }

    private void handleDataToRedis(DateTime date, String deviceCode, String notifyType) {
        // 准备Redis相关数据
        String formattedDate = DateUtil.format(date, "yyyyMMdd");
        String targetKeyV = "REPEATED_OFFLINE_ONLINE_" + formattedDate;
        // 从Redis获取现有数据
        String hgetData = jedis.hget(targetKeyV, deviceCode);
        JSONObject deviceStatusCount = new JSONObject();
        // 解析现有数据或初始化
        if (StringUtils.isNotBlank(hgetData)) {
            try {
                deviceStatusCount = JSONObject.parseObject(hgetData);
            } catch (Exception e) {
                log.error("解析Redis数据失败, deviceCode: {}, data: {}", deviceCode, hgetData, e);
                deviceStatusCount = new JSONObject();
            }
        }
        // 初始化计数器
        if (!deviceStatusCount.containsKey("online")) {
            deviceStatusCount.put("online", 0);
        }
        if (!deviceStatusCount.containsKey("offline")) {
            deviceStatusCount.put("offline", 0);
        }
        // 根据通知类型更新计数器
        if (Objects.equals(notifyType, ONLINE)) {
            deviceStatusCount.put("online", deviceStatusCount.getIntValue("online") + 1);
        } else if (Objects.equals(notifyType, OFFLLINE)) {
            deviceStatusCount.put("offline", deviceStatusCount.getIntValue("offline") + 1);
        }
        // 将更新后的数据写回Redis
        try {
            jedis.hset(targetKeyV, deviceCode, deviceStatusCount.toJSONString());
            // 设置过期时间（72小时过期）
            jedis.expire(targetKeyV, 72 * 60 * 60); // 24小时过期
        } catch (Exception e) {
            log.error("更新Redis数据失败, deviceCode: {}, targetKey: {}", deviceCode, targetKeyV, e);
        }
    }


    private void initDeviceCache() {
        try (Connection connection = rdsUtils.getConnection()) {
            // 创建 QueryRunner 对象
            QueryRunner queryRunner = new QueryRunner();
            // 设置分页参数
            int pageSize = 3000;
            int currentPage = 1;
            while (true) {
                log.info("加载{}缓存 {} {}", tableName5, currentPage, pageSize);
                int offset = (currentPage - 1) * pageSize;
                List<DeviceMoDO> query = queryRunner.query(connection, rds5SQL, new BeanListHandler<>(DeviceMoDO.class), pageSize, offset);
                if (query == null || query.isEmpty()) {
                    break;
                }
                for (DeviceMoDO deviceContainerDO : query) {
                    iotpDevicecaches.put(deviceContainerDO.getDeviceCode(), deviceContainerDO.getProductId());
                }
                currentPage++;
            }
            log.info("{} 缓存加载完成,大小为{}", tableName5,iotpDevicecaches.size());
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


    private void initProductCache() {
        try (Connection connection = rdsUtils.getConnection()) {
            // 创建 QueryRunner 对象
            QueryRunner queryRunner = new QueryRunner();
            // 设置分页参数
            int pageSize = 3000;
            int currentPage = 1;
            while (true) {
                log.info("加载{}缓存 {} {}", tableName6, currentPage, pageSize);
                int offset = (currentPage - 1) * pageSize;
                List<ProductDO> query = queryRunner.query(connection, rds6SQL, new BeanListHandler<>(ProductDO.class), pageSize, offset);
                if (query == null || query.isEmpty()) {
                    break;
                }
                for (ProductDO productDO : query) {
                    iotProductcaches.put(productDO.getProductId(), productDO.getProductType());
                }
                currentPage++;
            }
            log.info("{} 缓存加载完成,缓存大小为{}", tableName6, iotProductcaches.size());
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    @Override
    public void close() throws Exception {
        if (rdsUtils != null) {
            rdsUtils.close();
        }
        if (jedis != null) {
            jedis.close();
        }
    }
}

